package com.bonc.jedisCluster.utils;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.commons.pool2.impl.GenericObjectPoolConfig;

import com.bonc.jedisCluster.connectionHandler.MyJedisSlotBasedConnectionHandler;
import com.bonc.jedisCluster.utils.RedisDataTypeDescription.RedisDataType;

import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisClusterCommand;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.ScanParams;
import redis.clients.jedis.ScanResult;
import redis.clients.jedis.params.SetParams;
import redis.clients.jedis.util.JedisClusterCRC16;

/**
 * 1. 通过优化数据结构或者使用hashtag避免多节点操作数据</br>
 * 2. 如果需要多节点操作数据，尽量减少网络通信的次数，使用批量操作命令或者pipeline</br>
 * 3. 注意slot迁移场景（ASK重定向和MOVED重定向）。</br>
 * 4. Ask重定向：</br>
 * 		单条操作没有影响，JedisCluster封装了迁移自动调用Asking的功能</br>
 * 		批量key操作（例如mget、mset等）会有bug，会有slot迁移的问题，用封装的PipelineHandler代替</br>
 * 		批量非key操作不会有问题（例如keys，scan等）
 * 未解决：scan时发生slots迁移的现象
 * @author WYB
 *
 */
public class JedisClusterExecutor extends JedisCluster {
	
	private static JedisClusterExecutor jedisClusterExecutor;
	private static final int defaultScanCounter = 10;
	
	/**
	 * 重新定义了connectionHandler，新增getJedisPoolFromSlot方法
	 * @param jedisClusterNode
	 * @param connectionTimeout
	 * @param soTimeout
	 * @param maxAttempts
	 * @param poolConfig
	 */
	private JedisClusterExecutor(Set<HostAndPort> jedisClusterNode, int connectionTimeout, int soTimeout,
	    int maxAttempts, final GenericObjectPoolConfig poolConfig) {
	  super(jedisClusterNode, connectionTimeout, soTimeout, maxAttempts, poolConfig);
	  // 由于JedisCluster中默认的connectionHandler缺少getJedisPoolFromSlot方法，所以自定义connectionHandler，新增了该方法
	  super.connectionHandler = new MyJedisSlotBasedConnectionHandler(jedisClusterNode, poolConfig, connectionTimeout, soTimeout);
	}
	
	private JedisClusterExecutor(Set<HostAndPort> jedisClusterNode, int connectionTimeout, int soTimeout,
	    int maxAttempts, String password, final GenericObjectPoolConfig poolConfig) {
	  super(jedisClusterNode, connectionTimeout, soTimeout, maxAttempts, password, poolConfig);
	  // 由于JedisCluster中默认的connectionHandler缺少getJedisPoolFromSlot方法，所以自定义connectionHandler，新增了该方法
	  super.connectionHandler = new MyJedisSlotBasedConnectionHandler(jedisClusterNode, poolConfig, connectionTimeout, soTimeout, password);
	}
	
	/**
	 * 由于每一个JedisClusterExecutor对象都把所有的slot和node对应jedisPool的信息存储在内存中，故创建JedisClusterExecutor对象采用单例模式
	 * @param jedisClusterNode
	 * @param connectionTimeout 连接超时
	 * @param soTimeout 读写超时
	 * @param maxAttempts 重试次数
	 * @param poolConfig 连接池参数
	 * @return
	 */
	public static JedisClusterExecutor getInstance(Set<HostAndPort> jedisClusterNode, int connectionTimeout, int soTimeout,
		    int maxAttempts, final GenericObjectPoolConfig poolConfig) {
		if (jedisClusterExecutor==null) {
			synchronized (JedisClusterExecutor.class) {
				if (jedisClusterExecutor==null) {
					jedisClusterExecutor = new JedisClusterExecutor(jedisClusterNode, connectionTimeout, soTimeout, maxAttempts, poolConfig);
				}
			}
		}
		return jedisClusterExecutor;
	}
	
	public static JedisClusterExecutor getInstance(Set<HostAndPort> jedisClusterNode, int connectionTimeout, int soTimeout,
			int maxAttempts, String password, final GenericObjectPoolConfig poolConfig) {
		if (jedisClusterExecutor==null) {
			synchronized (JedisClusterExecutor.class) {
				if (jedisClusterExecutor==null) {
					jedisClusterExecutor = new JedisClusterExecutor(jedisClusterNode, connectionTimeout, soTimeout, maxAttempts, password, poolConfig);
				}
			}
		}
		return jedisClusterExecutor;
	}
	
	public MyJedisSlotBasedConnectionHandler getConnectionHandler() {
		return (MyJedisSlotBasedConnectionHandler) this.connectionHandler;
	}
	
	/**
	 * 负责scan操作
	 * 实现execute方法，负责对scan的结果集进行操作
	 * @author WYB
	 *
	 * @param <T>
	 */
	private abstract class ScanExecutor<T> {
		private Jedis jedis;
		private String cursor = "0";
		private ScanParams scanParams;
		/**
		 * 抽象方法，对scan的结果集进行的操作，在使用时自定义实现
		 * @param jedis
		 * @param keyList
		 * @return
		 */
		public abstract T execute(Jedis jedis, List<String> keyList);
		
		/**
		 * @param jedis 具体scan节点的jedis连接
		 * @param scanCounter 每次scan的个数
		 */
		public ScanExecutor(Jedis jedis, int scanCounter) {
			this.jedis = jedis;
			this.scanParams = new ScanParams().count(scanCounter);
		}
		
		// 没有模式的全匹配key
		public List<T> scan() {
			return scan(this.scanParams);
		}
		
		// 根据pattern定义的模式来匹配key
		public List<T> scan(String pattern) {
			return scan(this.scanParams.match(pattern));
		}
		
		private List<T> scan(ScanParams scanParams) {
			List<T> resultList = new ArrayList<T>();
			while (true) {
				ScanResult<String> scanResult = jedis.scan(cursor, scanParams);
				List<String> keyList = scanResult.getResult();
				if (keyList!=null && keyList.size()>0) {
					// 调用execute的具体实现，对scan的结果集进行操作
					T result = execute(this.jedis, keyList);
					resultList.add(result);
				}
				cursor = scanResult.getCursor();
				if ("0".equals(cursor)) {
					break;
				}
			}
			return resultList;
		}
	}
	
	/**
	 * 负责hscan操作
	 * @author WYB
	 *
	 */
	private class HScanExecutor {
		private Jedis jedis;
		private String cursor = "0";
		private ScanParams scanParams;
		
		public HScanExecutor(Jedis jedis, int scanCounter) {
			this.jedis = jedis;
			this.scanParams = new ScanParams().count(scanCounter);
		}
		public Map<String, String> hscan(String key) {
			return hscan(key, this.scanParams);
		}
		
		public Map<String, String> hscan(String key, ScanParams scanParams) {
			Map<String, String> resultMap = new HashMap<String, String>();
			while (true) {
				ScanResult<Entry<String, String>> hscanResult = jedis.hscan(key, cursor, scanParams);
				List<Entry<String, String>> mapList = hscanResult.getResult();
				for (Entry<String, String> entry : mapList) {
					resultMap.put(entry.getKey(), entry.getValue());
				}
				cursor = hscanResult.getCursor();
				if ("0".equals(cursor)) {
					break;
				}
			}
			return resultMap;
		}
	}
	
	public interface ExecuteOnNode<T> {
		/**
		 * 抽象方法，在一个节点上执行的多个key的操作
		 * @param connection 每个节点的连接
		 * @param keysOnNode 每个节点要处理的keys
		 * @return
		 */
		T executeOnNode(Jedis connection, String... keysOnNode);
	}
	
	/**
	 * 多Slots的处理器
	 * 提供一个将keys映射到JedisPool的方法
	 * @author WYB
	 *
	 */
	private abstract class MultiSlotsHandler<T> implements ExecuteOnNode<T> {

		/**
		 * 获得keys的JedisPool映射
		 * @param keys 需要映射的keys（可能分布在集群的多个节点）
		 * @return JedisPool到keys的映射，map的key是JedisPool，map的value是JedisPool对应的key的集合
		 */
		Map<JedisPool, List<String>> getKeysPool(String... keys) {
			Map<JedisPool, List<String>> keysPool = new HashMap<JedisPool, List<String>>();
			for (String key: keys) {
				// 将key映射到slot
				int slot = JedisClusterCRC16.getSlot(key);
				// 将slot映射到jedisPool，使用的是自定义MyJedisSlotBasedConnectionHandler的getJedisPoolFromSlot方法
				JedisPool jedisPool = ((MyJedisSlotBasedConnectionHandler) connectionHandler).getJedisPoolFromSlot(slot);
				if (keysPool.containsKey(jedisPool)) {
					keysPool.get(jedisPool).add(key);
				} else {
					List<String> list = new ArrayList<String>();
					list.add(key);
					keysPool.put(jedisPool, list);
				}
			}
			return keysPool;
		}
	}
	
	/**
	 * 处理多slots的get方法
	 * @author WYB
	 *
	 * @param <T>
	 */
	private abstract class MultiSlotsGetHandler<T> extends MultiSlotsHandler<T> {
		
		public List<T> run(String... keys) {
			List<T> resultList = new ArrayList<T>();
			Map<JedisPool, List<String>> keysPool = this.getKeysPool(keys);
			// 在每个节点的Jedis连接上处理属于这个节点的keys
			for (List<String> lessKeys : keysPool.values()) {
				// lessKeys都在一个节点上
				String[] lessKeysArr = lessKeys.toArray(new String[lessKeys.size()]);
				T values = new JedisClusterCommand<T>(connectionHandler, maxAttempts) {

					@Override
					public T execute(Jedis connection) {
						return executeOnNode(connection, lessKeysArr);
					}
					// 通过lessKeysArr[0]可以路由到这个key的节点上，由于lessKeys都在一个节点上，所以可以在得到的连接上，处理这些keys
				}.run(lessKeysArr[0]);
				resultList.add(values);
			}
			return resultList;
		}
		
		public void runAsync(String... keys) {
			Map<JedisPool, List<String>> keysPool = this.getKeysPool(keys);
			for (List<String> lessKeys : keysPool.values()) {
				// lessKeys都在一个节点上
				String[] lessKeysArr = lessKeys.toArray(new String[lessKeys.size()]);
				new Thread(new Runnable() {
					
					@Override
					public void run() {
						// TODO Auto-generated method stub
						Void values = new JedisClusterCommand<Void>(connectionHandler, maxAttempts) {
							
							@Override
							public Void execute(Jedis connection) {
								executeOnNode(connection, lessKeysArr);
								return null;
							}
							// 通过lessKeysArr[0]可以路由到这个key的节点上，由于lessKeys都在一个节点上，所以可以在得到的连接上，处理这些keys
						}.run(lessKeysArr[0]);
					}
				}).start();
			}
		}
	}
	
	/**
	 * 处理多slots的set方法
	 * @author WYB
	 *
	 * @param <T>
	 */
	private abstract class MultiSlotsSetHandler<T> extends MultiSlotsHandler {
		
		public abstract T executeOnNode(Jedis connection, String... keysvaluesOnNode);
		
		public List<T> run(String... keysvalues) {
			List<T> resultList = new ArrayList<T>();
			String[] keys = new String[keysvalues.length / 2];
			Map<String, String> keysvaluesMap = new HashMap<String, String>();

		    for (int keyIdx = 0; keyIdx < keys.length; keyIdx++) {
		      keys[keyIdx] = keysvalues[keyIdx * 2];
		      keysvaluesMap.put(keysvalues[keyIdx * 2], keysvalues[keyIdx * 2 + 1]);
		    }
			Map<JedisPool, List<String>> keysPool = this.getKeysPool(keys);
			for (List<String> lessKeys : keysPool.values()) {
				List<String> keysvaluesList = new ArrayList<String>();
				for (String lessKey: lessKeys) {
					keysvaluesList.add(lessKey);
					keysvaluesList.add(keysvaluesMap.get(lessKey));
				}
				String[] keysvaluesArr = keysvaluesList.toArray(new String[keysvaluesList.size()]);
				T values = new JedisClusterCommand<T>(connectionHandler, maxAttempts) {
					
					@Override
					public T execute(Jedis connection) {
						return executeOnNode(connection, keysvaluesArr);
					}
				}.run(keysvaluesArr[0]);
				resultList.add(values);
			}
			return resultList;
		}
	}

	/**
	 * 获得JedisCluster元数据的类</br>
	 * 调用的是jedisCluster中的方法，如果元数据更新（例如虚拟槽的rebalance），下次调用将能够感知到更新
	 * @author WYB
	 *
	 */
	public class Metadata {
		
		private JedisCluster jedisCluster;
		private Map<String, List<Integer>> nodeSlotRangeMap;
		
		/**
		 * 不允许创建对象
		 * @param jedisCluster
		 */
		public Metadata(JedisCluster jedisCluster) {
			this.jedisCluster = jedisCluster;
		}
		
		/**
		 * 得到Cluster包含的nodes和对应的JedisPool
		 * @return
		 */
		public Map<String, JedisPool> getClusterNodes() {
			return jedisCluster.getClusterNodes();
		}
		
		/**
		 * 得到Node对应的SlotRange
		 * @return
		 */
		public Map<String, List<Integer>> getNodeSlotRangeMap() {
			Map<String, List<Integer>> result = new HashMap<String, List<Integer>>();
			Map<String, JedisPool> nodes = getClusterNodes();
			for (Entry<String, JedisPool> node : nodes.entrySet()) {
				result.put(node.getKey(), new ArrayList<Integer>(2));
				Jedis jedis = null;
				try {
					jedis = node.getValue().getResource();
					List<Object> slots = jedis.clusterSlots();
					for (Object slotInfoObj : slots) {
						List<Object> slotInfo = (List<Object>) slotInfoObj;
						int slotStart = ((Long) slotInfo.get(0)).intValue();
						int slotEnd = ((Long) slotInfo.get(1)).intValue();
						result.get(node.getKey()).add(slotStart);
						result.get(node.getKey()).add(slotEnd);
					}
				} catch (Exception e) {
					e.printStackTrace();
					continue;
				} finally {
					if (jedis!=null) {
						jedis.close();
					}
				}
			}
			return result;
		}
		
		/**
		 * 得到key对应的slot
		 * @param key
		 * @return
		 */
		public int getSlot(String key) {
			return JedisClusterCRC16.getSlot(key);
		}
		
		/**
		 * 得到key对应的node
		 * @param key
		 * @return
		 */
		public String getNode(String key) {
			return getNode(getSlot(key));
		}
		
		/**
		 * 得到slot对应的node
		 * @param slot
		 * @return
		 */
		public String getNode(int slot) {
			Map<String, List<Integer>> nodeSlotRangeMap = getNodeSlotRangeMap();
			for(Map.Entry<String, List<Integer>> e: nodeSlotRangeMap.entrySet()) {
				List<Integer> slotRange = e.getValue();
				String node = e.getKey();
				if (slot >= slotRange.get(0) && slot <= slotRange.get(1)) {
					return node;
				}
			}
			return null;
		}
		
		/**
		 * 得到node对应的空闲连接的个数</br>
		 * 在外部分布式系统使用JedisCluster对象时，会很有用
		 * @return
		 */
		public Map<String, Integer> getNodeIdleConnectionMap() {
			Map<String, Integer> result = new HashMap<String, Integer>();
			Map<String, JedisPool> clusterNodes = getClusterNodes();
			for(String node: clusterNodes.keySet()) {
				result.put(node, clusterNodes.get(node).getNumIdle());
			}
			return result;
		}
	}
	
	public Metadata getMetadata() {
		return new Metadata(this);
	}
	
	/**
	 * 通过replication中的信息，来判断该节点是否是主节点
	 * @param jedis
	 * @return
	 */
	public static boolean isMaster(Jedis jedis) {
		String[] data = jedis.info("replication").split("\r\n");
		for (String line : data) {
			if ("role:master".equals(line.trim())) {
				return true;
			}
		}
		return false;
	}

	/**
	 * 如果是Jedis直连，则关闭Jedis连接
	 * 如果是JedisPool申请的连接，则归还连接
	 * @param connection
	 */
	private void releaseConnection(Jedis connection) {
		if (connection != null) {
			connection.close();
		}
	}

	public List<Long> delAllNodesByPattern(String pattern) {
		return delAllNodesByPattern(pattern, defaultScanCounter);
	}
	
	/**
	 * 与keysAllNodes原理类似
	 * @param pattern
	 * @param scanCounter
	 * @return
	 */
	public List<Long> delAllNodesByPattern(String pattern, int scanCounter) {
		Map<String, JedisPool> nodes = this.getClusterNodes();
		List<Long> results = new ArrayList<Long>();
		for (Entry<String, JedisPool> node : nodes.entrySet()) {
			Jedis jedis = null;
			try {
				jedis = node.getValue().getResource();
				if (!isMaster(jedis)) {
					continue;
				}
				results.addAll(new ScanExecutor<Long>(jedis, scanCounter) {

					@Override
					public Long execute(Jedis jedis, List<String> keyList) {
						Long del = jedis.del(keyList.toArray(new String[keyList.size()]));
						return del;
					}
				}.scan(pattern));
			} finally {
				releaseConnection(jedis);
			}
		}
		return results;
	}
	
	/**
	 * 原生的mget不支持跨slot的keys的查询</br>
	 * 新版本的api支持跨slot的keys查询</br>
	 * 先将keys全部映射到slots，再将slots映射到JedisPool，这样就能得到JedisPool对应的keys的映射，再遍历JedisPool处理该连接上的所有keys，得到整个集群的mget结果
	 * @param keys
	 * @return
	 */
	public List<List<String>> mgetAllNodes(String... keys) {
		return new MultiSlotsGetHandler<List<String>>() {

			@Override
			public List<String> executeOnNode(Jedis connection, String... keysOnNode) {
				return new PipelineHandler(connection, connectionHandler).mget(keysOnNode);
			}
		}.run(keys);
	}
	
	
	public List<List<String>> msetAllNodes(String... keysvalues) {
		return new MultiSlotsSetHandler<List<String>>() {

			@Override
			public List<String> executeOnNode(Jedis connection, String... keysvaluesOnNode) {
				return new PipelineHandler(connection, connectionHandler).mset(keysvaluesOnNode);
			}
		}.run(keysvalues);
	}
	
	/**
	 * 在集群的所有主节点上执行flushAll操作
	 * @return
	 */
	public List<String> flushAllNodes() {
		List<String> results = new ArrayList<String>();
		Map<String, JedisPool> nodes = this.getClusterNodes();
		for (Entry<String, JedisPool> node : nodes.entrySet()) {
			Jedis jedis = null;
			try {
				jedis = node.getValue().getResource();
				if (!isMaster(jedis)) {
					continue;
				}
				results.add(jedis.flushAll());
			} finally {
				releaseConnection(jedis);
			}
		}
		return results;
	}
	
	public List<Set<String>> keysAllNodes(String pattern) {
		return keysAllNodes(pattern, defaultScanCounter);
	}
	
	/**
	 * 用scan代替传统的keys方法，提高效率
	 * 支持集群操作keys
	 * @param pattern
	 * @param scanCounter
	 * @return
	 */
	public List<Set<String>> keysAllNodes(String pattern, int scanCounter) {
		Map<String, JedisPool> nodes = this.getClusterNodes();
		List<Set<String>> results = new ArrayList<Set<String>>();
		// 遍历集群所有节点的连接池，在每一个节点的Jedis连接上进行scan遍历，对scan的结果执行execute操作
		for (Entry<String, JedisPool> node : nodes.entrySet()) {
			Jedis jedis = null;
			try {
				jedis = node.getValue().getResource();
				// 只在主节点来操作
				if (!isMaster(jedis)) {
					continue;
				}
				results.addAll(new ScanExecutor<Set<String>>(jedis, scanCounter) {

					@Override
					public Set<String> execute(Jedis jedis, List<String> keyList) {
						return new HashSet<String>(keyList);
					}
				}.scan(pattern));
			} finally {
				// 需要自己实现对连接池中申请的空闲连接的归还
				releaseConnection(jedis);
			}
		}
		return results;
	}
	
	@Override
	public Map<String, String> hgetAll(String key) {
		return hgetAll(key, defaultScanCounter);
	}
	
	/**
	 * 用hscan渐进式遍历，代替传统的hgetall，提高性能
	 * @param key
	 * @param scanCounter
	 * @return
	 */
	public Map<String, String> hgetAll(String key, int scanCounter) {
		// JedisClusterCommand封装了key到slot和slot到connection的路由过程，以及获取连接和释放连接的过程，不需要自行实现
		return new JedisClusterCommand<Map<String, String>>(connectionHandler, maxAttempts) {
			@Override
			public Map<String, String> execute(Jedis connection) {
				return new HScanExecutor(connection, scanCounter) {
				}.hscan(key);
			}
		}.run(key);
	}

	/**
	 * 单机锁（上锁），锁住的是客户端
	 * 同一时间（在锁的过期时间内）只有一个客户端能够获得锁，锁存在集群中的某一个节点上
	 * 适用于避免多个客户端访问统一资源、做重复的工作等场景
	 * @param lockKey 锁的主键
	 * @param lockSecond 锁过期时间，过期会自动释放锁，避免死锁
	 * @return
	 */
	public String lock(String lockKey, int lockSecond) {
		String randomValue = String.valueOf(Math.random());
		String result = this.set(lockKey, randomValue, new SetParams().nx().ex(lockSecond));
		if ("OK".equals(result)) {
			return randomValue;
		}
		return null;
	}

	/**
	 * 单机锁（解锁）
	 * @param lockKey 锁的主键
	 * @param lockVal 自己创建的锁的值，避免错误的释放非本人创建的锁
	 */
	public void unlock(String lockKey, String lockVal) {
		if (lockVal.equals(this.get(lockKey))) {
			this.del(lockKey);
		}
	}

	public void executeOnNodeWithKeysAsync(ExecuteOnNode execOnNode, String... keys) {
		new MultiSlotsGetHandler<Void>() {
	
			@Override
			public Void executeOnNode(Jedis connection, String... keysOnNode) {
				execOnNode.executeOnNode(connection, keysOnNode);
				return null;
			}
		}.runAsync(keys);
	}
}
